package com.syher.seconds.kill.provider.service.impl;

import com.alibaba.dubbo.config.annotation.Service;
import com.google.common.base.Strings;
import com.syher.seconds.kill.constant.RedisKeysConstant;
import com.syher.seconds.kill.pojo.Stock;
import com.syher.seconds.kill.pojo.StockOrder;
import com.syher.seconds.kill.provider.mapper.OrderMapper;
import com.syher.seconds.kill.service.IOrderService;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;

@Repository
@Service(interfaceClass = IOrderService.class, dynamic = true, proxy = "jdk")
public class OrderService extends BaseServiceImpl<StockOrder, OrderMapper> implements OrderMapper {

    @Autowired
    StockService stockService;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Autowired
    private KafkaProducer<String, Object> kafkaProducer;

    @Value("${kafka.topic}")
    private String kafkaTopic;

    @Transactional
    public int createWrongOrder(int sid) throws Exception {
        //校验库存
        Stock stock = checkStock(sid);

        //扣库存
        saleStock(stock);

        //创建订单
        int id = createOrder(stock);

        return id;
    }

    @Transactional
    public int createOptimisticOrder(int sid) throws Exception {

        //校验库存
        Stock stock = checkStock(sid);

        //乐观锁更新库存
        saleStockOptimistic(stock);

        //创建订单
        int id = createOrder(stock);

        return id;
    }

    public int createOptimisticOrderUseRedis(int sid) throws Exception {
        //检验库存，从 Redis 获取
        Stock stock = checkStockByRedis(sid);

        //乐观锁更新库存 以及更新 Redis
        saleStockOptimisticByRedis(stock);

        //创建订单
        int id = createOrder(stock);
        return id;
    }

    public void createOptimisticOrderUseKafka(int sid) throws Exception {
        //检验库存，从 Redis 获取
        Stock stock = checkStockByRedis(sid);

        //利用 Kafka 创建订单
        kafkaProducer.send(new ProducerRecord(kafkaTopic, stock));
    }

    public void createOptimisticOrderUseRedisAndKafka(int sid) throws Exception {
        Stock stock = checkStockByRedis(sid);

        //乐观锁更新库存 以及更新 Redis
        saleStockOptimisticByRedis(stock);

        //创建订单
        createOrder(stock);
    }

    private Stock checkStockByRedis(int sid) throws Exception {
        String count = redisTemplate.opsForValue().get(RedisKeysConstant.STOCK_COUNT + sid);
        String sale = redisTemplate.opsForValue().get(RedisKeysConstant.STOCK_SALE + sid);
        String version = redisTemplate.opsForValue().get(RedisKeysConstant.STOCK_VERSION + sid);
        String name = redisTemplate.opsForValue().get(RedisKeysConstant.STOCK_NAME + sid);
        Stock stock;
        if (Strings.isNullOrEmpty(count) || Strings.isNullOrEmpty(sale)) {
            stock = checkStock(sid);
        } else if (Integer.parseInt(count) <= Integer.parseInt(sale)) {
            throw new RuntimeException("库存不足 Redis currentCount=" + sale);
        } else {
            stock = new Stock();
            stock.setId(sid);
            stock.setName(name);
            stock.setCount(Integer.parseInt(count));
            stock.setSale(Integer.parseInt(sale));
            stock.setVersion(Integer.parseInt(version));
        }

        return stock;
    }

    /**
     * 乐观锁更新数据库 还要更新 Redis
     *
     * @param stock
     */
    private void saleStockOptimisticByRedis(Stock stock) {
        saleStockOptimistic(stock);

        //自增
        redisTemplate.opsForValue().set(RedisKeysConstant.STOCK_SALE + stock.getId(), String.valueOf(stock.getSale() + 1));
        redisTemplate.opsForValue().set(RedisKeysConstant.STOCK_COUNT + stock.getId(), String.valueOf(stock.getCount()));
        redisTemplate.opsForValue().set(RedisKeysConstant.STOCK_VERSION + stock.getId(), String.valueOf(stock.getVersion() + 1));
        redisTemplate.opsForValue().set(RedisKeysConstant.STOCK_NAME + stock.getId(), stock.getName());
    }

    private void saleStockOptimistic(Stock stock) {
        int count = stockService.updateStockByOptimistic(stock);
        if (count == 0) {
            throw new RuntimeException("并发更新库存失败");
        }
    }

    private Stock checkStock(int sid) {
        Stock stock = stockService.selectByPrimaryKey(sid);
        if (stock.getSale() >= stock.getCount()) {
            throw new RuntimeException("库存不足");
        }
        return stock;
    }

    private int saleStock(Stock stock) {
        stock.setSale(stock.getSale() + 1);
        return stockService.updateByPrimaryKeySelective(stock);
    }

    private int createOrder(Stock stock) {
        StockOrder order = new StockOrder();
        order.setSid(stock.getId());
        order.setName(stock.getName());
        return getMapper().insertSelective(order);
    }
}
